package com.atguigu.day07;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

// 将两条流的历史数据进行内联结查询
// SELECT * FROM StreamA INNER JOIN StreamB WHERE StreamA.key = StreamB.key;
public class Example6 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Tuple2<String, Integer>> StreamA = env
                .fromElements(
                        Tuple2.of("a", 1),
                        Tuple2.of("b", 2),
                        Tuple2.of("a", 3),
                        Tuple2.of("b", 4)
                );

        DataStreamSource<Tuple2<String, Integer>> StreamB = env
                .fromElements(
                        Tuple2.of("a", 5),
                        Tuple2.of("b", 6),
                        Tuple2.of("a", 7),
                        Tuple2.of("b", 8)
                );

        // (a, 1) => (a, 5)
        // (a, 1) => (a, 7)
        // (a, 3) => (a, 5)
        // (a, 3) => (a, 7)
        // ...
        StreamA.keyBy(r -> r.f0)
                .connect(StreamB.keyBy(r -> r.f0))
                .process(new InnerJoin())
                .print();

        env.execute();
    }

    // 由于两条流都进行了keyBy
    // 所以进入到CoProcessFunction的数据拥有相同的key
    // 我们将来自两条流的相同key的数据放在了同一个逻辑分区中处理
    // 每个逻辑分区都有一个CoProcessFunction
    public static class InnerJoin extends
            CoProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String> {
        private ListState<Tuple2<String, Integer>> StreamAHistoryData;
        private ListState<Tuple2<String, Integer>> StreamBHistoryData;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            StreamAHistoryData = getRuntimeContext().getListState(
                    new ListStateDescriptor<Tuple2<String, Integer>>(
                            "StreamA-History-Data",
                            Types.TUPLE(Types.STRING, Types.INT)
                    )
            );
            StreamBHistoryData = getRuntimeContext().getListState(
                    new ListStateDescriptor<Tuple2<String, Integer>>(
                            "StreamB-History-Data",
                            Types.TUPLE(Types.STRING, Types.INT)
                    )
            );
        }

        @Override
        public void processElement1(Tuple2<String, Integer> value, Context ctx, Collector<String> out) throws Exception {
            // 处理的是来自第一条流的数据
            // 保存到历史数据
            StreamAHistoryData.add(value);
            // 将当前数据和B流的历史数据做join
            for (Tuple2<String, Integer> e : StreamBHistoryData.get()) {
                out.collect(value + " => " + e);
            }
        }

        @Override
        public void processElement2(Tuple2<String, Integer> value, Context ctx, Collector<String> out) throws Exception {
            // 处理的是来自第二条流的数据
            // 保存到历史数据
            StreamBHistoryData.add(value);
            // 将当前数据和A流的历史数据做join
            for (Tuple2<String, Integer> e : StreamAHistoryData.get()) {
                out.collect(e + " => " + value);
            }
        }
    }
}
